package com.petshop.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.petshop.mapper.VoucherOrderMapper;
import com.petshop.pojo.VoucherOrder;
import com.petshop.service.SeckillVoucherService;
import com.petshop.service.VoucherOrderService;
import com.petshop.utils.RedisIdWorker;
import com.petshop.utils.Result;
import com.petshop.utils.UserHolder;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author 20289
 * @description 针对表【tb_voucher_order】的数据库操作Service实现
 * @createDate 2025-03-31 14:32:09
 */
@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder>
        implements VoucherOrderService {

    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
    static {
        SECKILL_SCRIPT = new DefaultRedisScript<>();
        SECKILL_SCRIPT.setLocation(new ClassPathResource("seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }
    @Autowired
    private SeckillVoucherService seckillVoucherService;
    @Autowired
    private RedisIdWorker redisIdWorker;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    //注入RedissonClient对象
    @Autowired
    private RedissonClient redissonClient;
    //创建一个单线程的线程池，当作新的线程执行获取订单信息和创建订单的动作
    private static final ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();
    //初始化阻塞队列，设置其大小为1024*1024
//    private final BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);
//    /**
//     * 子线程从阻塞队列中取出来信息并异步下单写入主数据库
//     * 创建一个内部类，来执行异步下单的业务逻辑，从阻塞队列取出信息写入数据库。
//     */
//    private class VoucherOrderHandler implements Runnable {
//        @Override
//        public void run() {
//            //不断取出队列中的信息
//            while (true) {
//                //1.获取队列订单信息
//                try {
//                    VoucherOrder voucherOrder = orderTasks.take();
//                    //2.创建订单
//                    handleVoucherOrder(voucherOrder);
//                } catch (Exception e) {
//                    log.error("处理订单异常", e);
//                }
//            }
//        }
//    }
    String queueName = "streams.orders";
    private class VoucherOrderHandler implements Runnable {
        @Override
        public void run() {
            //不断取出队列中的信息
            while (true) {
                try {
                    //1.获取消息队列里的订单信息，XREAD group g1 消费者名字 count 1 [Block 2000 阻塞两秒] STREAMS streams.order >
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create(queueName, ReadOffset.lastConsumed())//代表>
                    );
                    //2.判断消息是否获取成功
                    if (list==null||list.isEmpty()) {
                        //2.1如果获取失败，说明没有下单消息 继续下次循环获取
                        continue;
                    }
                    //3.如果获取成功，就
                    //3.1解析消息队列中的信息
                        //因为我们设置的lua脚本里放入队列的就是键值对形式的值
                        //所以在这里被封装成map，String是消息id
                    MapRecord<String, Object, Object> record = list.get(0);
                    //得到我们传入的 'userId', userId,'voucherId', voucherId,'id',orderId
                    Map<Object, Object> value = record.getValue();
                    //转换成优惠券订单对象
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                    System.out.println("------"+voucherOrder);
                    //下单
                    handleVoucherOrder(voucherOrder);
                    //4.ACK确认 SACK streams.orders g1 id
                    stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
                } catch (Exception e) {
                    log.error("处理订单异常", e);
                    //如果抛出异常说明消息没有发送ACK确认，因此需要去pending-list取消息
                    try {
                        handlePendingList();
                    } catch (InterruptedException ex) {
                        log.error(ex.getMessage(), ex);
                    }
                }
            }
        }
    }

    private void handlePendingList() throws InterruptedException {
        while (true) {
            try {
                //1.获取消息队列里的订单信息，XREAD group g1 消费者名字 count 1  STREAMS streams.order 0
                List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1),
                        StreamOffset.create(queueName, ReadOffset.from("0"))//代表>
                );
                //2.判断消息是否获取成功
                if (list==null||list.isEmpty()) {
                    //2.1如果获取失败，说明pending-list没有异常消息 继续下次循环获取
                    break;
                }
                //3.如果获取成功，就
                //3.1解析消息队列中的信息
                //因为我们设置的lua脚本里放入队列的就是键值对形式的值
                //所以在这里被封装成map，String是消息id
                MapRecord<String, Object, Object> record = list.get(0);
                //得到我们传入的 'userId', userId,'voucherId', voucherId,'id',orderId
                Map<Object, Object> value = record.getValue();
                //转换成优惠券订单对象
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                //下单
                handleVoucherOrder(voucherOrder);
                //4.ACK确认 SACK streams.orders g1 id
                stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
            } catch (Exception e) {
                log.error("处理pending-list订单异常", e);
                Thread.sleep(20);
            }
        }
    }

    //这个内部类是在项目启动时初始化的，所以要加入注解@PostConstruct
    @PostConstruct
    private void init() {
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
    }

    /**
     * 主线程执行库存充足判断和是否一人一单。如果用户下单成功，
     * 存入信息到阻塞队列里，以供子线程使用信息创建订单。
     * 下单秒杀券
     *
     * @param voucherId
     * @return
     */
    private VoucherOrderService currentProxy;
    @Override
    public Result seckillVoucher(Long voucherId) {
        //获取用户
        Long userId = Long.valueOf(UserHolder.getUser().getUserId());
        //设置订单id
        Long voucherOrderId = redisIdWorker.nextId("voucherOrder");
        //1.执行lua脚本
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(),
                voucherId.toString(), userId.toString(),voucherOrderId.toString()
                //更改后，脚本不仅要判断购买资格，
                //还要发送下单信息到stream消息队列，也就不需要再用以前的阻塞队列了
        );
        //2.判断结果是否为0
        if (result.intValue() != 0) {
            //2.1.不是0就代表没有购买资格
            return Result.error(result == 1 ? "库存不足！" : "不能重复下单！");
        }
        //在主线程里获取代理对象,以便开启事务
        currentProxy = (VoucherOrderService) AopContext.currentProxy();
        //3.返回订单id
        return Result.ok(voucherOrderId);
    }
//    @Override
//    public Result seckillVoucher(Long voucherId) {
//        //获取用户
//        Long userId = Long.valueOf(UserHolder.getUser().getUserId());
//        //1.执行lua脚本
//        Long result = stringRedisTemplate.execute(
//                SECKILL_SCRIPT,
//                Collections.emptyList(),
//                voucherId.toString(), userId.toString()
//        );
//        //2.判断结果是否为0
//        if (result.intValue() != 0) {
//            //2.1.不是0就代表没有购买资格
//            return Result.error(result == 1 ? "库存不足！" : "不能重复下单！");
//        }
//        //2.2.是0代表有购买资格，将下单的信息保存至阻塞队列，以便后续根据队列异步实现下单
//        //创建优惠券订单
//        VoucherOrder voucherOrder = new VoucherOrder();
//        //设置订单id
//        long voucherOrderId = redisIdWorker.nextId("voucherOrder");
//        voucherOrder.setId(voucherOrderId);
//        //设置下单用户id
//        voucherOrder.setUserId(userId);
//        //设置优惠券id
//        voucherOrder.setVoucherId(voucherId);
//        // 2.3创建阻塞队列，放入上边的信息
//        orderTasks.add(voucherOrder);
//        //在主线程里获取代理对象,以便开启事务
//        currentProxy = (VoucherOrderService) AopContext.currentProxy();
//        //3.返回订单id
//        return Result.ok(voucherOrderId);
//    }

    /**
     * 再加一层锁保证并发安全
     * 调用创建订单的方法createVoucherOrder
     *
     * @param voucherOrder
     */
    private void handleVoucherOrder(@NotNull VoucherOrder voucherOrder) {
        //1.获取用户id
        Long userId = voucherOrder.getUserId();
        //2.创建锁对象
        RLock lock = redissonClient.getLock("lock:order:" + userId);
        //3.获取锁
        boolean isLock = lock.tryLock();
        //4.判断获取锁是否成功
        if (!isLock) {
            //获取锁失败
            log.error("不可以重复下单！");
        }
        try {
            currentProxy.createVoucherOrder(voucherOrder);
        } finally {
            //释放锁
            lock.unlock();
        }
    }

    @Transactional //添加事务，以便遇到问题及时回滚
    public void createVoucherOrder(VoucherOrder voucherOrder) {
        //5.充足就扣减库存
        boolean isOK = seckillVoucherService.update()
                .setSql("stock = stock -1")
                .eq("voucher_id", voucherOrder.getVoucherId())
//                .eq("stock", voucher.getStock()) //添加CAS乐观锁
                .gt("stock", 0) //改进乐观锁，stock>0即可修改数据了
                .update();
        if (!isOK) {
            //扣减失败
            log.error("库存不足！");
        }
        //6.订单信息写入数据库
        save(voucherOrder);
    }


}




